<?php

namespace SinoPHP\BbdBoss\Library;

use InvalidArgumentException;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Channel\AMQPChannel;

/**
 * Queue
 */
class Queue {

    const BASIC_QUEUE = 'BasicQueue';

    /**
     * Queue instance
     *
     * @var Queue
     */
    protected static $instance;

    /**
     * MQ Config
     *
     * @var array
     */
    protected static $config = array();

    /**
     * Queue name
     *
     * @var string
     */
    protected $queue;

    /**
     * Exchange name
     *
     * @var string
     */
    protected $exchange = 'TaskRunner';

    /**
     * Queue connection
     *
     * @var AMQPStreamConnection
     */
    private $connection;

    /**
     * Queue channel
     *
     * @var AMQPChannel
     */
    private $channel;

    protected function __construct($queue, array $config = array()) {
        $config = array_merge(self::$config, $config);
        $this->connection = new AMQPStreamConnection($config['host'], $config['port'], $config['user'], $config['pass'], $config['vhost']);

        $this->channel = $this->connection->channel();

        $this->queue = $queue;
    }

    /**
     * Configure
     *
     * @param array $config
     */
    public static function setConfig(array $config) {
        self::$config = array_merge(self::$config, $config);
    }

    /**
     * Connect MQ server
     *
     * @param string $queue
     * @param array  $config
     * @return Queue
     */
    public static function connect($queue = self::BASIC_QUEUE, array $config = array()) {
        if (!isset(self::$instance[$queue])) {
            self::$instance[$queue] = new self($queue, $config);
        }

        return self::$instance[$queue];
    }

    /**
     * Publish a message to MQ
     *
     * @param string  $worker
     * @param array   $params
     * @param boolean $ack
     * @param integer $maxTries
     */
    public function publish($worker, array $params = [], $ack = true, $maxTries = 1) {
        if (strpos($worker, '\\') !== false) {
            $workerName = explode('\\', $worker);
            $workerName = array_pop($workerName);
        } else {
            $workerName = $worker;
            $worker = '\\SinoPHP\\TaskRunner\\Consumer\\' . $worker;
        }

        $data = array(
            'name' => $workerName,
            'time' => time(),
            'worker' => $worker,
            'tries' => 0,
            'maxTries' => $maxTries > 1 ? $maxTries : $maxTries,
            'ack' => !empty($ack) ? true : false,
            'params' => !empty($params) ? $params : array(),
        );

        $this->publishMessage($data);
    }

    /**
     * Batch publish messages to MQ
     *
     * @param array $messages
     */
    public function batchPublish(array $messages) {
        $messageList = array();
        foreach ($messages as $msg) {
            if (empty($msg['worker'])) {
                throw new InvalidArgumentException('Miss param: worker');
            }
            if (strpos($msg['worker'], '\\') !== false) {
                $workerName = explode('\\', $msg['worker']);
                $workerName = array_pop($workerName);
            } else {
                $workerName = $msg['worker'];
                $msg['worker'] = '\\SinoPHP\\TaskRunner\\Consumer\\' . $msg['worker'];
            }

            $messageList[] = array(
                'name' => $workerName,
                'time' => time(),
                'worker' => $msg['worker'],
                'tries' => 0,
                'maxTries' => empty($msg['maxTries']) ? 1 : $msg['maxTries'],
                'ack' => !empty($msg['ack']) ? true : false,
                'params' => !empty($msg['params']) ? $msg['params'] : array(),
            );
        }

        $this->batchPublishMessage($messageList);
    }

    /**
     * Close queue connecting
     */
    public function close() {
        $this->channel->close();
        $this->connection->close();
    }

    /**
     * Publish a message
     *
     * @param  array $message
     */
    protected function publishMessage($message) {
        // name: $queue
        // passive: false
        // durable: true // the queue will survive server restarts
        // exclusive: false // the queue can be accessed in other channels
        // auto_delete: false //the queue won't be deleted once the channel is closed
        $this->channel->queue_declare($this->queue, false, true, false, false);

        // name: $exchange
        // type: direct
        // passive: false
        // durable: true // the exchange will survive server restarts
        // auto_delete: false //the exchange won't be deleted once the channel is closed.
        $this->channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $this->channel->queue_bind($this->queue, $this->exchange);

        // delivery_mode:
        // AMQPMessage::DELIVERY_MODE_PERSISTENT, make message persistent, so it is not lost if server crashes or quits
        $properties = array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT);
        $msgObj = new AMQPMessage(json_encode($message), $properties);

        // message
        // exchange
        // routing key (queue)
        $this->channel->basic_publish($msgObj, '', $this->queue);
    }

    /**
     * Batch publish messages
     *
     * @param array $messageList
     */
    protected function batchPublishMessage(array $messageList) {
        // name: $queue
        // passive: false
        // durable: true // the queue will survive server restarts
        // exclusive: false // the queue can be accessed in other channels
        // auto_delete: false //the queue won't be deleted once the channel is closed
        $this->channel->queue_declare($this->queue, false, true, false, false);

        // name: $exchange
        // type: direct
        // passive: false
        // durable: true // the exchange will survive server restarts
        // auto_delete: false //the exchange won't be deleted once the channel is closed.
        $this->channel->exchange_declare($this->exchange, 'direct', false, true, false);
        $this->channel->queue_bind($this->queue, $this->exchange);

        // delivery_mode:
        // AMQPMessage::DELIVERY_MODE_PERSISTENT, make message persistent, so it is not lost if server crashes or quits
        $properties = array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT);

        foreach ($messageList as $message) {
            $msgObj = new AMQPMessage(json_encode($message), $properties);
            // message
            // exchange
            // routing key (queue)
            $this->channel->batch_basic_publish($msgObj, '', $this->queue);
        }

        $this->channel->publish_batch();
    }

}
